package com.smsc.headend.task.engine.service.task.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.lang.Tuple;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import cn.hutool.json.JSONUtil;
import com.smsc.headend.common.enums.DeviceType;
import com.smsc.headend.common.enums.errorcode.TaskExecuteErrorCode;
import com.smsc.headend.common.enums.errorcode.TaskHandlerErrorCode;
import com.smsc.headend.common.exception.AbstractUDISException;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.data.dto.TaskDataDTO;
import com.smsc.headend.module.data.dto.TaskPacketDTO;
import com.smsc.headend.module.protocol.DeviceData;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.*;
import com.smsc.headend.module.task.entity.DeviceTask;
import com.smsc.headend.module.task.enums.SubTaskStatus;
import com.smsc.headend.module.task.enums.TaskStatus;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.module.task.proto.RetryTaskDtoProto;
import com.smsc.headend.module.task.proto.TaskProto;
import com.smsc.headend.module.task.proto.TaskResultProto;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.exception.UDISDeviceRespException;
import com.smsc.headend.task.engine.exception.UDISTaskExecutorException;
import com.smsc.headend.task.engine.exception.UDISTaskHandlerException;
import com.smsc.headend.task.engine.executor.AtomicTaskFactory;
import com.smsc.headend.task.engine.executor.TaskExecutor;
import com.smsc.headend.task.engine.executor.concurrent.CheckTaskExecutorRunner;
import com.smsc.headend.task.engine.handler.TaskHandler;
import com.smsc.headend.task.engine.handler.TaskResultHandlerFactory;
import com.smsc.headend.task.engine.manager.OdrRequestManager;
import com.smsc.headend.task.engine.po.CollectTask;
import com.smsc.headend.task.engine.po.TaskEndStatusPo;
import com.smsc.headend.task.engine.service.CosemXmlService;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.asset.AssetService;
import com.smsc.headend.task.engine.service.task.DeviceTaskService;
import com.smsc.headend.task.engine.service.task.DeviceTaskSubService;
import com.smsc.headend.task.engine.service.task.TaskDataService;
import com.smsc.headend.task.engine.service.task.TaskService;
import io.micrometer.core.instrument.Counter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;
import java.util.stream.Stream;


@Service
@Slf4j
public class TaskServiceImpl implements TaskService {

    @Autowired
    AtomicTaskFactory atomicTaskFactory;
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    CosemXmlService cosemXmlService;
    @Autowired
    TaskResultHandlerFactory taskResultHandlerFactory;
    @Autowired
    DeviceTaskService deviceTaskService;
    @Autowired
    DeviceTaskSubService deviceTaskSubService;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    TaskDataService taskDataService;
    @Autowired
    RedisUtils redisUtils;
    @Autowired
    FeignAssetManagementService assetManagementService;
    @Autowired
    @Qualifier("taskRequestThreadPool")
    ThreadPoolExecutor taskThreadPool;

    @Autowired
    @Qualifier("taskResponseThreadPool")
    ThreadPoolExecutor taskSendThreadPool;
    @Autowired
    TaskService taskService;
    @Autowired
    OdrRequestManager odrRequestManager;

    @Autowired
    @Qualifier("taskStopWatcher")
    ScheduledExecutorService taskStopWatcher;

    @Autowired
    @Qualifier("taskUTGInbound")
    Counter utgTaskInboundCounter;

    @Autowired
    @Qualifier("taskUTGOutbound")
    Counter utgTaskOutboundCounter;

    @Autowired
    AssetService assetService;


    @Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    ScheduledFuture<?> checkTaskPool = null;

    private Map<Long, Set<Tuple>> comProcessingTask = new ConcurrentHashMap<>();

    private Map<Long, ReentrantLock> comProcessLockMap = new HashMap<>();

    LinkedBlockingQueue<Runnable> rejectedTask = new LinkedBlockingQueue();

    private synchronized ReentrantLock getComLock(Long comId) {
        if (comProcessLockMap.containsKey(comId)) {
            return comProcessLockMap.get(comId);
        }
        ReentrantLock lock = new ReentrantLock();
        comProcessLockMap.put(comId, lock);
        return lock;
    }

    @Override
    public boolean updateTask(Task task) {
        return taskDAO.updateTask(task);
    }

    @Override
    public void sendTaskToGenerator(Task task) {
        if (!this.taskValidCheck(task)) {
            log.info("taskInvalidBeforeSendToUTG,taskNo:{},taskType:{}", task.getTaskNo(), task.getTaskType());
            return;
        }
        if (!TaskType.AutoReCollectProfile.toString().equalsIgnoreCase(task.getTaskType())
                && task.getEndTime() != null
                && task.getEndTime() != 0
                && task.getEndTime() <= SystemClock.now() / 1000) {
            log.info("taskExpired,taskNo:{},taskType:{}", task.getTaskNo(), task.getTaskType());
            return;
        }
        log.info("sendTaskToUTG: taskNo:{}, taskType:{}", task.getTaskNo(), task.getTaskType());
        kafkaTemplate.send(TaskKafkaTopic.TASK_GEN_FROM_OTHER, Convert.toStr(task.getComId()), ProtoBufferUtil.toProtoBuffer(task, TaskProto.Task.class).toByteArray());
    }

    private boolean taskValidCheck(Task task) {
        if (CollectionUtils.isEmpty(task.getSubTaskGroups()) || task.getComId() == null) {
            return false;
        }

        if (task.getSubTaskGroups().stream().filter(g -> g.getGroupId() == null || CollectionUtils.isEmpty(g.getSubTasks())).findAny().isPresent()) {
            return false;
        }

        if (task.getSubTaskGroups().stream().filter(g -> g.getSubTasks().stream().filter(this::atomicTaskInvalid).findAny().isPresent()).findAny().isPresent()) {
            return false;
        }
        return true;
    }

    private boolean atomicTaskInvalid(AtomicTask atomicTask) {
        if (atomicTask.getOperationId() == null) {
            return true;
        }

        if (StringUtils.isEmpty(atomicTask.getAtomicTaskNo())) {
            return true;
        }
        return false;
    }

    @Override
    public Task getTaskByNo(String taskNo) {
        return taskDAO.getTaskByNo(taskNo);
    }

    public CollectTask updatePreviousCollectTask(Task oldTask, Task collectionTask, Long startTime, Long endTime, CollectTask collectTask) {
        Map<String, Object> prevParam = JSONUtil.toBean(oldTask.getParamJson(), Map.class);
        Long prevStartTime = Convert.toLong(prevParam.get("profileStartTime"));
        Long prevEndTime = Convert.toLong(prevParam.get("profileEndTime"));
        log.debug("{}, {}, {}", startTime, endTime, oldTask.getTaskNo());
        if (prevEndTime < endTime) {
            prevParam.put("profileEndTime", endTime);
            collectTask.setEndTime(endTime);
            log.debug("collect task update profileEndTime: {}, {} -> {}", oldTask.getTaskNo(), prevEndTime, endTime);
        }
        if (prevStartTime > startTime) {
            prevParam.put("profileStartTime", startTime);
            collectTask.setStartTime(startTime);
            log.debug("collect task update profileStartTime: {}, {} -> {}", oldTask.getTaskNo(), prevStartTime, startTime);
        }
        oldTask.setParamJson(JSONUtil.toJsonStr(prevParam));
        oldTask.setEndTime(collectionTask.getEndTime());
        taskDAO.updateTask(oldTask);
        log.debug("collect task update: taskNo={},endTime={}", oldTask.getTaskNo(), oldTask.getEndTime());
        return collectTask;
    }


    @Override
    public void interruptComTask(Long comId) {
        Integer processingPriority = taskDAO.hashGetProcessingTerminalQueue(String.valueOf(comId));
        if (processingPriority == null) {
            return;
        }
        String terminalQueueId = taskDAO.getTerminalTaskListKey(comId, processingPriority);
        Task processingTask = taskDAO.terminalTaskQueueGet(terminalQueueId);
        log.info("task_execute: task interrupt,comId={}, taskNo={}", comId, processingTask != null ? processingTask.getTaskNo() : null);
        if (processingTask != null) {
            taskDAO.hashDelProcessingTerminalQueue(terminalQueueId);
            deviceTaskService.sendTaskStatusUpdateMessage(processingTask, processingTask.getMessageId(), TaskStatus.Suspending);
        }
    }


    @Override
    public TaskEndStatusPo doStartTask(Task task) {
        if (task == null || !taskValidCheck(task)) {
            return null;
        }
        utgTaskInboundCounter.increment();
        log.debug("TaskQueue start execute: comId={}, taskNo={}", task.getComId(), task.getTaskNo());
        deviceTaskService.sendTaskStatusUpdateMessage(task, task.getMessageId(), TaskStatus.Processing);

        this.addTaskProcessingInfo(task);
        if (isNeedAtomicTaskEnter(task)) {
            task.getSubTaskGroups().stream().map(SubTaskGroup::getSubTasks).flatMap(x -> x.stream()).
                    forEach(atomicTask -> {
                        this.pushAtomicTask(atomicTask);
                    });
        }
        odrRequestManager.taskStart(task);
        return this.emitAtomicTask(task, null);
    }

    private synchronized void addTaskProcessingInfo(Task task) {
        ReentrantLock comProcessLock = getComLock(task.getComId());
        comProcessLock.lock();
        try {
            if (comProcessingTask.containsKey(task.getComId())) {
                comProcessingTask.get(task.getComId()).add(new Tuple(task.getMeterId(), task.getTaskNo(), task.getPriority()));
            } else {
                HashSet<Tuple> set = new HashSet<>();
                set.add(new Tuple(task.getMeterId(), task.getTaskNo(), task.getPriority()));
                comProcessingTask.put(task.getComId(), set);
            }
        } catch (Exception e) {

        } finally {
            comProcessLock.unlock();
        }
    }

    private void removeTaskProcessingInfo(Task task) {
        Set<Tuple> tuples = comProcessingTask.get(task.getComId());
        if (CollectionUtils.isEmpty(tuples)) {
            log.error("TaskNoProcessingInfo,comId:{},meterId:{},taskType:{},taskNo:{}", task.getComId(), task.getMeterId(), task.getTaskType(), task.getTaskNo());
            return;
        }
        List<Tuple> priorityTasks = tuples.stream().filter(t -> t.get(2).equals(task.getPriority())).collect(Collectors.toList());
        if (!CollectionUtils.isEmpty(priorityTasks)) {
            ReentrantLock comProcessLock = getComLock(task.getComId());
            comProcessLock.lock();
            try {
                Tuple tuple = tuples.stream().filter(t -> t.get(1).equals(task.getTaskNo())).findFirst().orElse(null);
                if (tuple == null) {
                    log.info("taskNoNotFoundInEngine,comId:{},Priority:{},backFromQueue,meterId:{},taskType:{},taskNo:{}", task.getComId(), task.getPriority(), task.getMeterId(), task.getTaskType(), task.getTaskNo());
                    return;
                }
                log.debug("found task:{}, tuple:{}", task.getTaskNo(), JSONUtil.toJsonStr(tuple));
                tuples.remove(tuple);
                if (tuples.size() > 0) {
                    log.info("QueueInEngineClearForJumpQueue:{}", JSONUtil.toJsonStr(tuples));
                    tuples.clear();
                }
            } catch (Exception e) {
                log.error("remove comProcessingTaskFailed", e);
            } finally {
                comProcessLock.unlock();
            }
        } else {
            log.error("noTaskInEngineInPriorityComId:{},meterId:{},taskType:{},taskNo:{}", task.getComId(), task.getMeterId(), task.getTaskType(), task.getTaskNo());
        }
    }

    private boolean isNeedAtomicTaskEnter(Task task) {
//        //非固件传输任务  沿用旧逻辑，统一都赛一边进queue
//        if (!task.getTaskType().equals(TaskType.FirmwareImageTransfer.toString())) {
//            return true;
//        }
        //固件传输任务  如果已经有atomicTaskQueue,不需要重复入队列
        if (!taskDAO.atomicTaskQueueExist(task.getTaskNo())) {
            return true;
        }
        return false;
    }


    public Task nextTask(String terminalQueueId) {
        String[] terminalInfo = terminalQueueId.split(":");
        String prefixTerminal = String.join(":", Arrays.copyOfRange(terminalInfo, 0, terminalInfo.length - 1));
        int p = 1;
        Task task = null;
        //get task from queue in priority order
        while (p <= Task.Priority.LOW.getPriorityCode() && task == null) {
            String tempTermQueueId = prefixTerminal + ":" + p;
            task = taskDAO.terminalTaskQueueGet(tempTermQueueId);
            p++;
        }
        return task;
    }

    /**
     * AtomicTask执行完成时的处理方法
     *
     * @param task
     * @param st
     * @param atomicTaskCompleted -1 --设备返回异常，需要异常处理判断是否继续  1.设备返回成功，继续执行整个任务流程 0.原子任务不移除，继续调用，在Executor中处理对应的不同帧  2.重新插入执行队列尾部
     * @param exception
     * @return
     * @throws InstantiationException
     * @throws IllegalAccessException
     */
    public TaskEndStatusPo callbackTask(Task task, AtomicTaskState st, int atomicTaskCompleted, AbstractUDISException exception) throws InstantiationException, IllegalAccessException {
        TaskEndStatusPo atomicCallBackResult = new TaskEndStatusPo();
        if (task == null) {
            return atomicCallBackResult;
        }
        //为0时 不删除当前atomicTask, 继续执行
        log.debug("task callback: taskNo={},atomicTaskNo={}, complete={}", task.getTaskNo(), st == null ? null : st.getAtomicTaskNo(), atomicTaskCompleted);
        String failedMessage = null;
        if (exception != null) {
            log.info("task callback: taskNo={},atomicTaskNo={}, atomicTaskCompleted={}, exception={}", task.getTaskNo(), st.getAtomicTaskNo(), atomicTaskCompleted, exception.getMessage());
            failedMessage = exception.getMessage();
        }
        if (atomicTaskCompleted != 0) {
            atomicTaskRemove(st, task.getPriority());
        }
        //得到优先级最高的任务
//        Task nextTask = this.nextTask(taskDAO.getTerminalTaskListKey(task));
        Boolean taskEndFlag = false;
        Boolean taskRetry = false;
        if (atomicTaskCompleted == -1) {
            atomicCallBackResult.setSuccess(false);
            taskEndFlag = handleFailAtomicTask(task, st, exception);
            if (taskEndFlag) {
                taskRetry = true;
            }
        } else if (atomicTaskCompleted == 0) {
            if (queueJump(task)) {
                return atomicCallBackResult.setTaskEnd(false);
            }
            log.info("sendNextAtomicTask: taskNo={},blockNum={},collectEndTv={}", st.getTaskNo(), st.getBlockNumber(), st.getCollectingEndTv());
            this.emitAtomicTask(task, null);
        }
        //atomic task成功
        else if (atomicTaskCompleted == 1) {
            taskEndFlag = handleSuccessAtomicTask(task, st);
        } else if (atomicTaskCompleted == 2) {
            //重入队尾的任务直接结束
            taskEndFlag = true;
            handleReEnterTask(task, st);
            atomicCallBackResult.setTaskReEntry(true);
        } else {
            log.error("AtomicTaskResponse: taskCompleted error, taskNo={}, atomicTaskNo={}, taskCompleted={}", task.getTaskNo(), st == null ? null : st.getAtomicTaskNo(), atomicTaskCompleted);
        }
        odrRequestManager.atomicTaskEnd(task, st.getAtomicTaskNo(), atomicTaskCompleted);
        atomicCallBackResult.setTaskEnd(taskEndFlag).setTaskRetry(taskRetry).setMessage(failedMessage);
        return atomicCallBackResult;
    }

    private void handleReEnterTask(Task task, AtomicTaskState st) {
//        Boolean taskEndFlag = false;
        //判断是不是TaskGroup完成
        SubTaskGroup subTaskGroup = null;
        for (SubTaskGroup g : task.getSubTaskGroups()) {
            if (g.getSubTasks().get(g.getSubTasks().size() - 1).getAtomicTaskNo().equalsIgnoreCase(st.getAtomicTaskNo())) {
                subTaskGroup = g;
                break;
            }
        }

        if (subTaskGroup != null) {
            //sub task group finish
            log.debug("task group finish");
            if (deviceTaskService.isTaskSubPersist(task)) {
                Integer sort = deviceTaskSubService.getAtomicTaskGroupSeq(task, st.getAtomicTaskNo());
                deviceTaskService.sendSubTaskMessageUpdate(task, sort, SubTaskStatus.Success.getCode());
            }
        }
    }

    private Boolean handleSuccessAtomicTask(Task task, AtomicTaskState st) throws InstantiationException, IllegalAccessException {
        Boolean taskEndFlag = false;
        //判断是不是TaskGroup完成
        SubTaskGroup subTaskGroup = null;
        for (SubTaskGroup g : task.getSubTaskGroups()) {
            if (g.getSubTasks().get(g.getSubTasks().size() - 1).getAtomicTaskNo().equalsIgnoreCase(st.getAtomicTaskNo())) {
                subTaskGroup = g;
                break;
            }
        }

        if (subTaskGroup != null) {
            //sub task group finish
            log.debug("task group finish");
            if (deviceTaskService.isTaskSubPersist(task)) {
                Integer sort = deviceTaskSubService.getAtomicTaskGroupSeq(task, st.getAtomicTaskNo());
                deviceTaskService.sendSubTaskMessageUpdate(task, sort, SubTaskStatus.Success.getCode());
            }
        }

        if (queueJump(task)) {
            return false;
        }
        //start next atomic task
        TaskEndStatusPo taskEndStatusPo = this.emitAtomicTask(task, st.getAtomicTaskNo());
        if (taskEndStatusPo == null || taskEndStatusPo.isTaskEnd()) {
            log.debug("task {}'s no more a-task", task.getTaskNo());
            taskEndFlag = true;
            sendSuccessMessageToAdapter(task);
        }
        return taskEndFlag;
    }

    private Boolean handleFailAtomicTask(Task task, AtomicTaskState st, AbstractUDISException exception) {
        Boolean taskEndFlag = false;
        if (st != null && deviceTaskService.isTaskSubPersist(task)) {
            Integer sort = deviceTaskSubService.getAtomicTaskGroupSeq(task, st.getAtomicTaskNo());
            deviceTaskService.sendSubTaskMessageUpdate(task, sort, SubTaskStatus.Failed.getCode());
        }
        if (!task.getFailStop()) {
            TaskEndStatusPo taskEndStatusPo = this.emitAtomicTask(task, null);
            if (taskEndStatusPo == null || taskEndStatusPo.isTaskEnd()) {
                taskEndFlag = true;
                sendFailMessageToAdaptor(task, exception == null ? "null" : exception.getMessage());
            }
        } else {
//            sendRetryMessage(task);
            taskEndFlag = true;
        }
        return taskEndFlag;
    }

    private void sendRetryMessage(Task task) {
        if (task.getRetryTimes() == null || task.getRetryTimes() == 0) {
            return;
        }
        Integer executedTimes = task.getExecutedTimes() == null ? 0 : task.getExecutedTimes();
        if (executedTimes >= task.getRetryTimes()) {
            log.error("task {} retry times used up", task.getTaskNo());
            sendReplyToAdapter(task);
            return;
        }
        if (task.getEndTime() != null && task.getEndTime() != 0 && task.getEndTime() < SystemClock.now() / 1000) {
            log.error("task expired {}", task.getTaskNo());
            sendReplyToAdapter(task);
            return;
        }
        log.info("retry task send to utg {}", task.getTaskNo());
        taskDAO.hashSetRetryTask(task);
        RetryTaskDto dto = new RetryTaskDto();
        dto.setTaskNo(task.getTaskNo());
        dto.setExecutedTimes(executedTimes + 1);
        dto.setNextExecutionTime(SystemClock.now() / 1000 + task.getRetryIntervalSeconds());
        kafkaTemplate.send(TaskKafkaTopic.TASK_RETRY_TO_UTG, ProtoBufferUtil.toProtoBuffer(dto, RetryTaskDtoProto.RetryTaskDto.class).toByteArray());
    }

    public Boolean queueJump(Task task) {
        Set<Tuple> tuples = this.comProcessingTask.get(task.getComId());
        if (tuples == null) {
            return false;
        }
        Tuple priorTaskInfo = tuples.stream().filter(t -> !t.get(1).equals(task.getTaskNo()) && Convert.toInt(t.get(2)) < task.getPriority()).findAny().orElse(null);
        if (priorTaskInfo != null) {
            log.info("jump in: comId={}, meterId={}, taskNo={}", task.getComId(), priorTaskInfo.get(0), priorTaskInfo.get(1));
            deviceTaskService.sendTaskStatusUpdateMessage(task, task.getMessageId(), TaskStatus.Suspending);
            return true;
        }
        return false;
    }

    private void sendSuccessMessageToAdapter(Task task) {
        if (isSendToAdapter(task.getTaskType())) {
            List<TaskDataDTO> taskDataDTOS = taskDAO.getAllTaskData(task.getTaskNo());
            TaskResult taskResult = new TaskResult();
            taskResult.setMessageId(task.getMessageId());
            taskResult.setTaskNo(task.getTaskNo());
            taskResult.setResultData(taskDataDTOS);
            taskResult.setParamJson(task.getParamJson());
            kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(taskResult).getBytes(Charset.forName("utf-8")));
        }
    }

    private boolean isSendToAdapter(String taskType) {
        if (StrUtil.isNotEmpty(taskType)) {
            return StrUtil.equalsAny(taskType, TaskType.UAA_Provision.toString(),
                    TaskType.UAA_MeterControl.toString());
        }
        return false;
    }

    private void sendFailMessageToAdaptor(Task task, String exceptionText) {
        // UAA失败返回
        if (isSendToAdapter(task.getTaskType())) {
            List<TaskDataDTO> taskDataDTOS = taskDAO.getAllTaskData(task.getTaskNo());
            TaskResult taskResult = new TaskResult();
            taskResult.setMessageId(task.getMessageId());
            taskResult.setTaskNo(task.getTaskNo());
            taskResult.setExceptionCause(exceptionText);
            taskResult.setResultData(taskDataDTOS);
            taskResult.setParamJson(task.getParamJson());
            kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(taskResult).getBytes(Charset.forName("utf-8")));
        }
    }

    private void endTask(Task task, TaskEndStatusPo taskEndStatusPo) {
        if (task == null || StringUtils.isEmpty(task.getTaskNo())) {
            return;
        }
        //current task pop out
        taskDAO.atomicTaskQueueDelete(task.getTaskNo());
        taskDAO.clearTaskProcessingAtomic(task.getTaskNo());
        removeTaskProcessingInfo(task);
        log.info("TaskEnd: comId:{},taskNo:{}", task.getComId(), task.getTaskNo());

        TaskResultProto.TaskResult.Builder builder = TaskResultProto.TaskResult.newBuilder();
        builder.setTaskNo(task.getTaskNo())
                .setComId(task.getComId())
                .setSuccess(taskEndStatusPo.isSuccess())
                .setMessageId(task.getMessageId())
                .setPriority(task.getPriority())
                .setMessage(Optional.ofNullable(taskEndStatusPo.getMessage()).orElse(""));
        TaskResultProto.TaskResult taskResult = builder.build();
        kafkaTemplate.send(TaskKafkaTopic.TASK_IN_UTE_END, Convert.toStr(task.getComId()), taskResult.toByteArray());
        utgTaskOutboundCounter.increment();
        try {
            postEndTask(task, taskEndStatusPo);
        } catch (Exception e) {
            log.error("post end failed, taskNo={}, msgId={}", task.getTaskNo(), task.getMessageId(), e);
        }
    }

    /**
     * 任务结束后的操作
     *
     * @param task
     * @param taskEndStatusPo
     */
    private void postEndTask(Task task, TaskEndStatusPo taskEndStatusPo) {
        Boolean updateTaskCount = true;
        if (taskEndStatusPo.isTaskRetry()) {
            sendRetryMessage(task);
            updateTaskCount = false;
        }
        if (deviceTaskService.isTaskSubPersist(task)) {
            deviceTaskService.clearEstimateCache(task.getComId(), task.getMeterId() == null ? -1 : task.getMeterId());
        }
        odrRequestManager.taskEnd(task, taskEndStatusPo.isSuccess());
        if (updateTaskCount) {
            deviceTaskService.sendTaskStatusUpdateMessage(task, task.getMessageId(), TaskStatus.Complete);
        }
    }

    @Override
    public void atomicTaskResponse(ConnectorTaskResult connectorTaskResult) throws IllegalAccessException, InstantiationException {
        String atomicTaskNo = connectorTaskResult.getAtomicTaskNo();
        AtomicTaskState st = taskDAO.hashGetAtomicTaskState(atomicTaskNo);
        if (st == null) {
            log.error("atomic task not exists {}", atomicTaskNo);
            return;
        }
        Task task = taskDAO.getTaskByNo(st.getTaskNo());
        log.info("taskResponse: comId={},taskNo={}", task != null ? task.getComId() : null, task != null ? task.getTaskNo() : null);
        //判断是表计返回错误,或者是获取下一个block,或者成功可以进入下一个任务
        int atomicTaskCompleted = -1;
        AbstractUDISException exception = null;
        try {
            atomicTaskCompleted = handleAtomicTask(connectorTaskResult, task, st);
        } catch (UDISTaskHandlerException | UDISDeviceRespException e) {
            log.error("UDIS inner exception {}", e.getMessage(), e);
            atomicTaskCompleted = -1;
            exception = e;
        } catch (Exception e) {
            log.info("{}", JSONUtil.toJsonPrettyStr(Stream.of(e.getStackTrace()).limit(10).map(p -> p.getClassName() + ":" + p.getLineNumber()).collect(Collectors.toList())));
            log.error("Unhandled error ", e);
            atomicTaskCompleted = -1;
            exception = new UDISTaskHandlerException(e.getMessage());
        }
//        redisUtils.unlockEasyRedisLock(st.getAtomicTaskNo(), lockKey);
        TaskEndStatusPo taskEndStatusPo = this.callbackTask(task, st, atomicTaskCompleted, exception);
        log.debug("task response: AtomicCallBackResult={}", taskEndStatusPo);
        this.handleTaskEndPo(taskEndStatusPo, task);
    }

    /**
     * remove atmoic task  from Queue
     *
     * @param st
     * @param priority
     */
    private void atomicTaskRemove(AtomicTaskState st, Integer priority) {
        taskDAO.atomicTaskQueuePop(st.getTaskNo());
        taskDAO.clearTaskProcessingAtomic(st.getTaskNo());
        taskDAO.hashDeleteAtomicTaskState(st.getAtomicTaskNo(), priority);
    }


    /**
     * @param connectorTaskResult
     * @param task
     * @param st
     * @return -1 -- Error, Interrupt  1 -- Complete  0- next block  2 --re-enter to task queue
     */
    private int handleAtomicTask(ConnectorTaskResult connectorTaskResult, Task task, AtomicTaskState st) throws UDISTaskHandlerException, UDISDeviceRespException {
        DeviceData cosemData = null;
        if (task == null) {
            return -1;
        }
        AtomicTask atomicTask = task.getSubTaskGroups().stream().flatMap(subTaskGroup -> subTaskGroup.getSubTasks().stream())
                .filter(atomicTask1 -> atomicTask1.getAtomicTaskNo().equalsIgnoreCase(st.getAtomicTaskNo())).findFirst().orElse(null);

        if (atomicTask == null) {
            AtomicTask firstAtomicTask = taskDAO.atomicTaskQueueGet(task.getTaskNo());
            if (firstAtomicTask == null || !firstAtomicTask.getAtomicTaskNo().equals(connectorTaskResult.getAtomicTaskNo())) {
                log.error("task execute：atomic task not found. comId={},taskNo={},atomicTaskNo={}", task.getComId(), task.getTaskNo(), st.getAtomicTaskNo());
                return -1;
            }
            atomicTask = firstAtomicTask;
        }

        TaskHandler taskHandler = null;
        try {
            taskHandler = taskResultHandlerFactory.getHandler(task, atomicTask, st.getHandlerClass());
            if (connectorTaskResult.getExecuteCode() == ConnectorTaskResult.ExecuteCode.fail.getCode()) {
                log.error("task execute： failed.comId={},taskType={},reason={}", task.getComId(), task.getTaskType(), connectorTaskResult.getResultMessage());
                sendFailMessageToAdaptor(task, connectorTaskResult.getResultMessage());
                return taskHandler.handleFailure(task, atomicTask, connectorTaskResult.getResultMessage());
            }

            if (StringUtils.isNotEmpty(connectorTaskResult.getResultData())) {
                try {
                    cosemData = cosemXmlService.handlerXML(task, atomicTask, connectorTaskResult.getResultData());
                } catch (Exception e) {
                    log.error("task execute: handler result xml failed, resultData={}", connectorTaskResult.getResultData());
                }
            }
            return taskHandler.handle(task, atomicTask, cosemData, connectorTaskResult.getResultMessage());
        } catch (UDISTaskHandlerException e) {
//            addODRMessage(task, atomicTask, e);
            if (UDISTaskExecutorException.ODR_INTERRUPT.equalsIgnoreCase(e.getMessage())) {
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Cancel.getCode());
            } else if (UDISTaskExecutorException.TASK_EXPIRED.equalsIgnoreCase(e.getMessage())) {
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Expired.getCode());
                log.debug("task execute: expired");
            } else {
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Failed.getCode());
            }
            if (taskHandler != null) {
                return taskHandler.handleFailure(task, atomicTask, e.getMessage());
            }
            throw e;
        } catch (Exception e) {
            log.error("Unexpected error", e);
            if (UDISTaskExecutorException.ODR_INTERRUPT.equalsIgnoreCase(e.getMessage())) {
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Cancel.getCode());
            } else if (UDISTaskExecutorException.TASK_EXPIRED.equalsIgnoreCase(e.getMessage())) {
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Expired.getCode());
                log.debug("task execute: expired");
            } else {
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Failed.getCode());
            }
            if (taskHandler != null) {
                return taskHandler.handleFailure(task, atomicTask, e.getMessage());
            }
        }
        throw new UDISTaskHandlerException(TaskHandlerErrorCode.TASK_HANDLER_EXECUTE_EXCEPTION.getErrorCode());
    }

    @Override
    public TaskEndStatusPo emitAtomicTask(Task task, String previousAtomicTaskNo) {
        return doEmitAtomicTask(task, 0, previousAtomicTaskNo);
    }

    private TaskEndStatusPo doEmitAtomicTask(Task task, int sendTimes, String previousAtomicTaskNo) {
        TaskEndStatusPo taskEndStatusPo = new TaskEndStatusPo();
        AtomicTask atomicTask = taskDAO.atomicTaskQueueGet(task.getTaskNo());
        if (atomicTask == null || StringUtils.isEmpty(atomicTask.getAtomicTaskNo())) {
            return null;
        }
        if (atomicTask.getAtomicTaskNo().equals(previousAtomicTaskNo)) {
            log.error("AtomicTaskNoDuplicated: taskNo:{},taskType:{},atomicTaskNo:{},subTasks:{}", task.getTaskNo(), task.getTaskType(), previousAtomicTaskNo, JSONUtil.toJsonStr(task.getSubTaskGroups()));
            return null;
        }
        //get redis lock
        Long lockKey = redisUtils.easyRedisLock(atomicTask.getAtomicTaskNo());
        if (lockKey != null) {
            //get operation meter network  type, ip address , operationType,  protocol, OBIS ，class id. send to connector
            try {
                preCheckTask(task);
                log.debug("task execute: emit taskNo={}", atomicTask);
                TaskExecutor taskExecutor = atomicTaskFactory.getTaskExecutor(atomicTask.getOperationId());
                if (taskExecutor == null) {
                    throw new UDISTaskExecutorException(TaskExecuteErrorCode.NO_EXECUTOR.toString());
                }
                AtomicTaskState atomicTaskState = taskExecutor.execute(task, atomicTask);
                redisUtils.unlockEasyRedisLock(atomicTask.getAtomicTaskNo(), lockKey);
                taskDAO.hashSetAtomicTaskState(atomicTask.getAtomicTaskNo(), task, sendTimes, atomicTaskState);
                taskDAO.setTaskProcessingAtomicTaskNo(task.getTaskNo(), atomicTask.getAtomicTaskNo());
                deviceTaskService.updateSubDeviceTaskStart(task, atomicTask);
                odrRequestManager.atomicTaskStart(task, atomicTask);
            } catch (UDISTaskExecutorException | Exception e) {
                try {
                    taskEndStatusPo = handleExecuteException(task, atomicTask, e);
                } catch (Exception e1) {
                    taskEndStatusPo.setTaskEnd(true);
                    taskEndStatusPo.setSuccess(false);
                    taskEndStatusPo.setMessage("execute fail");
                    log.error("handle failure exception", e1);
                }
                odrRequestManager.atomicTaskStartFailed(task, atomicTask);
            } finally {
                redisUtils.unlockEasyRedisLock(atomicTask.getAtomicTaskNo(), lockKey);
            }
        } else {
            log.info("task execute: lock fail, comId={}, atomicNo={}, sendTimes={}", task.getComId(), atomicTask.getAtomicTaskNo(), sendTimes);
            return null;
        }

        return taskEndStatusPo;
    }

    private TaskEndStatusPo handleExecuteException(Task task, AtomicTask atomicTask, Throwable e) {
        TaskEndStatusPo taskEndStatusPo = new TaskEndStatusPo().setTaskEnd(true);
        if (e != null && TaskExecuteErrorCode.RECOLLECT_TASK_EXECUTED_RECENTLY.getErrorCode().equalsIgnoreCase(e.getMessage())) {
            log.debug("task execute: a-Task failed,comId={}, taskNo={},atomicNo={},failMessage={}", task.getComId(), task.getTaskNo(), atomicTask.getAtomicTaskNo(), e.getMessage());
        } else {
            log.error("task execute: a-Task failed,comId={}, taskNo={},taskType={},failMessage={}", task.getComId(), task.getTaskNo(), task.getTaskType(), e == null ? null : e.getMessage(), e);
        }
        TaskExecutor taskExecutor = atomicTaskFactory.getTaskExecutor(atomicTask.getOperationId());
        String failedMessage =  e == null ? "execute fail" : e.getMessage();
        if (taskExecutor != null) {
            Boolean isTaskEnd = taskExecutor.handleFailure(task, atomicTask, failedMessage);
            taskEndStatusPo.setTaskEnd(isTaskEnd);
        }
        sendFailMessageToAdaptor(task, failedMessage);
        if (e != null) {
            log.trace("error com={}", task.getComId(), e);
            if (UDISTaskExecutorException.ODR_INTERRUPT.equalsIgnoreCase(e.getMessage())) {
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Cancel.getCode());
            } else if (UDISTaskExecutorException.TASK_EXPIRED.equalsIgnoreCase(e.getMessage())) {
                log.debug("task execute: expired");
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Expired.getCode());
            } else {
                deviceTaskService.updateRequestDeviceTask(task, SubTaskStatus.Failed.getCode());
            }
        }
        taskEndStatusPo.setSuccess(false);
        taskEndStatusPo.setMessage(failedMessage);
        return taskEndStatusPo;
    }

    private void preCheckTask(Task task) throws UDISTaskExecutorException {
        if (task == null) {
            return;
        }

        if (!task.getTaskType().equalsIgnoreCase(TaskType.AutoReCollectProfile.toString())
                && task.getEndTime() != null && task.getEndTime() != 0 && task.getEndTime() <= SystemClock.now() / 1000) {
            throw new UDISTaskExecutorException(UDISTaskExecutorException.TASK_EXPIRED + task.getEndTime());
        }

        if (task.getIsODR() && StringUtils.isNotEmpty(task.getMessageId())) {
            Long queryTime = taskDataService.getMessageIdTime(task.getMessageId());
            //没有轮询记录。odr任务结束
            if (queryTime == null || SystemClock.now() - queryTime > 60000) {
                throw new UDISTaskExecutorException(UDISTaskExecutorException.ODR_INTERRUPT);
            }
        }

//        ComEntity comEntity = assetService.getComById(task.getComId());
//        if (comEntity == null) {
//            throw new UDISTaskExecutorException(UDISTaskExecutorException.COM_NOT_EXSITS);
//        }
//        if (isComEntityOffline(comEntity)) {
//            throw new UDISTaskExecutorException(UDISTaskExecutorException.COM_STATUS_EXCEPTION);
//        }
    }

    private boolean isComEntityOffline(ComEntity com) {
        if (com.getComMode() == ComEntity.CommunicationMode.CLIENT.getCode()) {
            return false;
        }
        Integer comStatus = com.getComStatus();
        return comStatus == ComEntity.CommunicationStatus.OFFLINE.getCode() || comStatus == ComEntity.CommunicationStatus.NEVER.getCode();
    }


    @Override
    public Set<String> getAtomicTaskProcessingList() {
        return taskDAO.hashGetAtomicTaskStateSet();
    }

    @Override
    public void pushAtomicTask(AtomicTask atomicTask) {
        taskDAO.atomicTaskQueueAdd(atomicTask);
    }

    @Override
    public void checkTimeOver(Set<String> processingList) throws InstantiationException, IllegalAccessException {
        for (String atomicTaskNo : processingList) {
            taskThreadPool.execute(() -> {
                taskService.checkAtomicTaskOverTime(atomicTaskNo);
            });
        }
    }

    @Override
    public int checkAtomicTaskOverTime(String atomicTaskNo) {
        AtomicTaskState st = taskDAO.hashGetAtomicTaskState(atomicTaskNo);
        if (st == null) {
            return 0;
        }
        Long currentTime = SystemClock.now() / 1000;
        if (currentTime - 120 > st.getLastTime()) {
            log.info("over time ,taskNo={}, lastTime={}", st.getTaskNo(), st.getLastTime());
            try {
                handleOverTime(st);
            } catch (Exception e) {
                log.error("handle time over fail");
                log.debug("faied excetion:", e);
            }
        }
        return 1;
    }

    private void handleOverTime(AtomicTaskState st) throws IllegalAccessException, InstantiationException {
        log.debug("over time remove task {}", st.getTaskNo());
        Task task = taskDAO.getTaskByNo(st.getTaskNo());
        //固件升级ImageTransfer超时时间延长为1小时
        if (task != null && task.getTaskType().equalsIgnoreCase(TaskType.FirmwareImageTransfer.toString()) && SystemClock.now() / 1000 - st.getLastTime() <= 3600) {
            log.info("Image Transfer Job waiting reconnect taskNo={} comId={} meterId={}", task.getTaskNo(), task.getComId(), task.getMeterId());
            return;
        }

        if (task != null) {
            log.info("task execute: over time task, taskNo={}, a-task={}", st.getTaskNo(), st.getAtomicTaskNo());
            ConnectorTaskResult connectorTaskResult = new ConnectorTaskResult();
            connectorTaskResult.setResultMessage(TaskExecuteErrorCode.TASK_WAITING_CON_RESPONSE_OVERTIME.getErrorText());
            connectorTaskResult.setAtomicTaskNo(st.getAtomicTaskNo());
            connectorTaskResult.setExecuteCode(ConnectorTaskResult.ExecuteCode.fail.getCode());
            this.atomicTaskResponse(connectorTaskResult);
        }

    }

    @Override
    public List<TaskDataDTO> taskDataPopAll(String taskNo) {
        return taskDAO.popAllTaskData(taskNo);
    }

    @Override
    public Task getTask(Long terminalId, Integer priority) {
        String key = taskDAO.getTerminalTaskListKey(terminalId, priority);
        return taskDAO.terminalTaskQueueGet(key);
    }

    @Override
    public String getProcessingAtomicTaskByTaskNo(String taskNo) {
        return taskDAO.getProcessingAtomicTaskByTaskNo(taskNo);
    }

    @Override
    public Long getDeviceId(Task task) {
        Long deviceId = task.getMeterId();
        if (task.getMeterId() == null) {
            ComEntity comEntity = assetService.getComById(task.getComId());
            deviceId = comEntity.getDeviceId();
        }
        return deviceId;
    }

    @Override
    public Long getDeviceType(Task task) {
        if (task.getMeterId() == null) {
            ComEntity comEntity = assetService.getComById(task.getComId());
            return comEntity.getComDeviceType();
        } else {
            return DeviceType.METER.getId();
        }
    }


    @Override
    public List<TaskPacketDTO> getTaskPacket(String taskNo) {
        return taskDAO.popAllTaskPacket(taskNo);
    }

    @Override
    public List<TaskPacketDTO> getTaskPacket(DeviceTask deviceTask) {
        List<TaskPacketDTO> taskPacketDTOS = taskDAO.popAllTaskPacket(deviceTask);
        log.info("deviceTask:taskNo:{} ,{}", deviceTask.getTaskNo(), JSONUtil.toJsonStr(taskPacketDTOS));
        return taskPacketDTOS;
    }

    @Override
    public void autoRecollectCheck() {
        log.info("auto recollect: removed after refactor");
    }

    @Override
    public void handleTaskEndPo(TaskEndStatusPo taskEndStatusPo, Task task) {
        if (taskEndStatusPo.isTaskEnd()) {
            this.endTask(task, taskEndStatusPo);
        }

        if (taskEndStatusPo.isTaskReEntry()) {
            task.setExecutedTimes(task.getExecutedTimes() + 1);
            this.sendTaskToGenerator(task);
        }
    }

    @Override
    public void saveRejectedTask(Runnable r) {
        this.rejectedTask.add(r);
    }

    @Override
    public synchronized void scheduleCheckExecutor() {
        if (checkTaskPool == null || checkTaskPool.isDone()) {
            log.info("willStartCheckState 3s Later");
            CheckTaskExecutorRunner checkTaskExecutorRunner = new CheckTaskExecutorRunner();
            SpringUtil.getApplicationContext().getAutowireCapableBeanFactory().autowireBean(checkTaskExecutorRunner);
            checkTaskPool = taskStopWatcher.schedule(checkTaskExecutorRunner, 3, TimeUnit.SECONDS);
        }
    }

    @Override
    public void taskExecutionPoolCheck() {
        long taskSendPoolCount = taskSendThreadPool.getQueue().size();
        long taskThreadPoolTaskCount = taskThreadPool.getQueue().size();
        long sendActiveCount = taskSendThreadPool.getActiveCount();
        long taskExecuteActiveCount = taskThreadPool.getActiveCount();
        log.info("CurrentPoolSize:{},{},active:{},{}", taskThreadPoolTaskCount, taskSendPoolCount, sendActiveCount, taskExecuteActiveCount);
        if (taskThreadPoolTaskCount < 500 && taskSendPoolCount < 500 && this.rejectedTask.isEmpty()) {
            Collection<MessageListenerContainer> allListenerContainers = kafkaListenerEndpointRegistry.getAllListenerContainers();
            MessageListenerContainer listenerContainer = allListenerContainers.stream()
                    .filter(listener -> Arrays.stream(listener.getContainerProperties().getTopics()).anyMatch(s -> TaskKafkaTopic.TASK_SEND_TO_UTE.equals(s))).findFirst().get();
            log.info("resumeConsumeTask!");
            redisUtils.del("udis.stop.schedule.generating");
            listenerContainer.resume();
            return;
        }
        if (!this.rejectedTask.isEmpty() && taskSendPoolCount < 500) {
            int size = this.rejectedTask.size();
            int enterQueueSize = size < 50 ? size : 50;
            log.info("startReExecuteRejectTask: size:{}", enterQueueSize);
            for (int i = 0; i < enterQueueSize; i++) {
                taskSendThreadPool.submit(this.rejectedTask.poll());
            }
        }
        CheckTaskExecutorRunner checkTaskExecutorRunner = new CheckTaskExecutorRunner();
        SpringUtil.getApplicationContext().getAutowireCapableBeanFactory().autowireBean(checkTaskExecutorRunner);
        checkTaskPool = taskStopWatcher.schedule(checkTaskExecutorRunner, 100, TimeUnit.MILLISECONDS);
    }


    private void sendReplyToAdapter(Task task) {
        log.info("send retry failed to adapter,taskNo:{}", task.getTaskNo());
        if (TaskType.UAA_MeterReads.name().equals(task.getTaskType())) {
            TaskResult taskResult = new TaskResult();
            taskResult.setTaskNo(task.getTaskNo());
            taskResult.setMessageId(task.getMessageId());
            taskResult.setParamJson(task.getParamJson());
            taskResult.setExceptionCause("retry time out");
            kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(task).getBytes(Charset.forName("utf-8")));
        }
    }
}
